package com.fs.kafka.test;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @author shuai.fang
 * @date：2017-10-12
 * @description：
 */
public class KafkaProducer {

	private String brokerList;
	private String serializerKeyClass;
	private String serializerValueClass;

	Producer<String, String> producer;

	public KafkaProducer(String brokerList, String serializerKeyClass, String serializerValueClass) {
		this.brokerList = brokerList;
		this.serializerKeyClass = serializerKeyClass;
		this.serializerValueClass = serializerValueClass;
	}

	public void init() {
		Properties properties = new Properties();
		properties.setProperty("bootstrap.servers", brokerList);
		properties.setProperty("key.serializer", serializerKeyClass);
		properties.setProperty("value.serializer", serializerValueClass);

		producer = new org.apache.kafka.clients.producer.KafkaProducer<String, String>(properties);
	}

	public void sendMessage(String topic, String message) {
		if (producer == null) {
			throw new RuntimeException("请先调用init方法进行初始化");
		}

		System.out.format("向 %s 发送消息 -> %s", topic, message);
		ProducerRecord<String, String>  messageRecord = new ProducerRecord<String, String>(topic, message);
		producer.send(messageRecord, new Callback() {
			@Override
			public void onCompletion(RecordMetadata metadata, Exception exception) {
				System.out.println(metadata);
				System.out.println(exception);
			}
		});

		try {
			TimeUnit.SECONDS.sleep(1);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
}
